package com.pw.study.flink.chapter3;

import com.pw.study.flink.entities.OrderEvent;
import com.pw.study.flink.entities.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

public class AppOrderAndTx {

    public static void main(String[] args) {
        dealApp();
    }

    private static void dealApp() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        SingleOutputStreamOperator<OrderEvent> order = env.readTextFile("data/file/OrderLog.csv").map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String line) throws Exception {
                String[] order = line.split(",");
                Long id = Long.valueOf(order[0]);
                String type = order[1];
                String txId = order[2];
                Long ts = Long.valueOf(order[3]);
                return new OrderEvent(id, type, txId, ts);
            }
        }).filter(x -> x.getEventType().equals("pay"));

        SingleOutputStreamOperator<TxEvent> txt = env.readTextFile("data/file/ReceiptLog.csv").map(new MapFunction<String, TxEvent>() {
            @Override
            public TxEvent map(String line) throws Exception {
                String[] txt = line.split(",");
                String txId = txt[0];
                String pay = txt[1];
                Long ts = Long.valueOf(txt[2]);
                return new TxEvent(txId, pay, ts);
            }
        });

        order.connect(txt).keyBy(OrderEvent::getTxId, TxEvent::getTxId).process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
            final Map<String, OrderEvent> mapOrder = new HashMap<>();
            final Map<String, TxEvent> mapTxt = new HashMap<>();

            @Override
            public void processElement1(OrderEvent orderEvent, Context ctx, Collector<String> out) throws Exception {
                String txId = orderEvent.getTxId();
                if (mapTxt.containsKey(txId)) {
                    out.collect("订单：" + orderEvent.getOrderId() + " 对账成功");
                } else {
                    mapOrder.put(txId, orderEvent);
                }
            }

            @Override
            public void processElement2(TxEvent txEvent, Context ctx, Collector<String> out) throws Exception {
                String txId = txEvent.getTxId();
                if (mapOrder.containsKey(txId)) {
                    out.collect("订单：" + mapOrder.get(txId).getOrderId() + " 对账成功");
                } else {
                    mapTxt.put(txId, txEvent);
                }

            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
